<?php

/**
 * Created by PhpStorm.
 * User: 554665488
 * Date: 2018-7-6
 * Time: 19:16
 * @description: 多进程模型
 * @Author: yfl
 * @QQ 554665488
 */
class process1
{
    protected $table;

    public function __construct()
    {
        $this->table = $this->create_table();
        $this->init();
        $this->signal();//信号监听
    }


    public function init()
    {
        $data = [
            '990979940@qq.com',
            '213123@qq.com',
            '457567657@qq.com'
        ];
        foreach ($data as $index => $datum) {
            $process = $this->create_process();
//            var_dump('创建de pid--' . $process->pid);
            $this->table->set($process->pid, ['data' => $datum]);
            $process->write($datum);
        }
        foreach ($this->table as $index => $item) {
//            var_dump($item);
        }
    }

    //创建进程
    protected function create_process()
    {
        $process = new Swoole\Process([$this, 'callback_function']);
        $process->start(); //启动子进程
//异步读取管道消息
        Swoole\Event::add($process->pipe, function () use ($process) {
            $res_data = $process->read(); //获得进程的执行任务的结果
            $res_data = json_decode($res_data, true);
            if ($res_data['status'] == false) ;
//            echo '子进程发送的内容:' . PHP_EOL;
//            var_dump($res_data);
            Swoole\Event::del($process->pipe);//删除事件
        });
        return $process;
    }

    public function callback_function($worker)
    {
        $data = $worker->read(); //获取数据执行逻辑
        $status = mt_rand(1, 3);
        if ($status == 1) {
//            throw  new Exception('任务执行异常');
        }
        $res = [
            'status' => true, 'data' => $data, 'msg' => '任务执行成功'
        ];
//        var_dump('回调的pid--'.$worker->pid);
        $worker->write(json_encode($res)); //写入到主进程
        $this->table->del($worker->pid); //执行成功 删除数据
    }

    //创建内存表格
    public function create_table()
    {
        $table = new Swoole\Table(1024);
        $table->column('attempts', Swoole\Table::TYPE_INT, 5);
        $table->column('data', Swoole\Table::TYPE_STRING, 64);
        $table->create();
        return $table;
    }

    //子进程退出 信号监听
    protected function signal()
    {
        Swoole\Process::signal(SIGCHLD, function ($sig) {
            //false 未非阻塞模式
            while ($res = Swoole\Process::wait(false)) {
//                var_dump($res);
                if ($res['code'] !== 0) { //状态码非0状态 异常退出
//                    echo '进行重新投递';
                    $pid = $res['pid'];
                    $rev_data = $this->table->get($pid);
                    $this->table->del($pid);
                    //判断重试次数
                    if ($rev_data['attempts'] == 0 || $rev_data['attempts'] < 2) {
//                        echo '重新投递';
                        $process = $this->create_process(); //重新拉起进程,会产生新的pid
                        $this->table->set($process->pid, ['data' => $rev_data['data'], 'attempts' => $rev_data['attempts'] + 1]); //新创建的进程
                        $process->write($rev_data['data']);
                        return;
                    }
                    //file_put_contents(); 记录日志|发送邮件|异常队列当中
                }
            }
        });
    }
}

new process1();